package com.atguigu.flink.exec1;

import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * Created by Smexy on 2023/1/29
 * 
 *      ①读数据
 *      ②把原始数据封装为计算的类型
 *      ③计算
 *      ④输出
 *              print
 */
public class Demo1_PV
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        
        //读数据
        env.readTextFile("data/UserBehavior.csv")
           .map(new MapFunction<String, UserBehavior>()
           {
               @Override
               public UserBehavior map(String line) throws Exception {
                   String[] words = line.split(",");
                  return new UserBehavior(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       Integer.valueOf(words[2]),
                       words[3],
                       Long.valueOf(words[4])
                   );
               }
           })
           //过滤出PV的数据
           .filter(ub -> "pv".equals(ub.getBehavior()))
           .map(ub -> Tuple2.of("pv",1)).returns(TUPLE(STRING,INT))
           .keyBy(t -> t.f0)
           .sum(1)
           .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        
    }
}
